package client

import (
	"GraduationProject/public/db"
	"GraduationProject/public/models"
	"GraduationProject/public/tools"
	"encoding/json"
	"github.com/nsqio/go-nsq"
	"log"
	"sync"
	"time"
)

var (
	nsqConsumers consumers
	SeedChannel  chan models.Record
)

func init() {
	SeedChannel = make(chan models.Record)
	nsqConsumers = consumers{
		Mutex:     sync.Mutex{},
		container: make(map[string]*nsq.Consumer),
	}
	go UpdateConsumers()
}

type consumers struct {
	sync.Mutex
	container map[string]*nsq.Consumer
}

func UpdateConsumers() {
	for {
		if db.DB == nil {
			time.Sleep(5 * time.Second)
			continue
		}
		var nsqSettings []models.NsqSetting
		db.DB.Where("worker_id = ? and status = ?",
			tools.Conf.Get("worker_id"), models.NsqSettingStatusRunning).
			Find(&nsqSettings)

		nKey := make(map[string]struct{})
		for _, n := range nsqSettings {
			_, ok := nsqConsumers.container[n.Topic]
			if !ok {
				// consumer 不存在
				if err := AddConsumer(n.Host, n.Topic); err != nil {
					log.Printf("UpdateConsumers AddConsumer error: %s\n", err.Error())
					continue
				}
			}
			nKey[n.Topic] = struct{}{}
		}

		for topic, _ := range nsqConsumers.container {
			if _, ok := nKey[topic]; !ok {
				DeleteConsumer(topic)
			}
		}

		time.Sleep(time.Second * 30)
	}
}

func AddConsumer(host, topic string) error {
	nsqConsumers.Lock()
	defer nsqConsumers.Unlock()
	config := nsq.NewConfig()
	consumer, err := nsq.NewConsumer(topic, models.DefaultNsqChannel, config)
	if err != nil {
		return err
	}
	consumer.AddHandler(nsq.HandlerFunc(func(message *nsq.Message) error {
		var r models.Record
		if err := json.Unmarshal(message.Body, &r); err != nil {
			log.Printf("nsq.HandlerFunc json.Unmarshal error: %s\n", err.Error())
			return err
		}
		//log.Printf("SeedChannel <- r: %#v\n", r)
		SeedChannel <- r
		return nil
	}))
	if err := consumer.ConnectToNSQD(host); err != nil {
		return err
	}
	log.Printf("consumer.ConnectToNSQD(%#v) success\n", host)

	nsqConsumers.container[topic] = consumer
	return nil
}

func DeleteConsumer(topic string) {
	nsqConsumers.Lock()
	defer nsqConsumers.Unlock()
	consumer := nsqConsumers.container[topic]
	if consumer != nil {
		consumer.Stop()
		delete(nsqConsumers.container, topic)
	}
}
